/*
 * =====================================================================================
 *
 *       Filename:  test_chatserver.cxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  04/28/2011 06:40:27 AM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Chen Ken (Ken), zlchen.ken@gmail.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef __GXX_EXPERIMENTAL_CXX0X__
#define __GXX_EXPERIMENTAL_CXX0X__
#endif 

#include <arpa/inet.h>
#include <set>
#include "cl_tcp_server.hxx"
#include "cl_tcp_client.hxx"
#include "cl_tcp_conn.hxx"
#include "cl_buffer.hxx"
#include "cl_mutex.hxx"
#include "cl_lock_guard.hxx"
#include "cl_signal_handler.hxx" 

using namespace std;
using namespace std::placeholders;
using namespace neo;

class SimpleCodec
{
public:
typedef function<void(const CLTcpConnSPtr&, 
                      const string& msg, 
                      const CLTimestamp& tim)> MsgCallback;
  explicit SimpleCodec(const MsgCallback& cb)
  	: m_msgCallback(cb)
  {
  }
  
  void handleMsg(const CLTcpConnSPtr& conn, CLBuffer& buf, const CLTimestamp& tim)
  {
  	while(buf.readableBytes() >= HEAD_SIZE)
  	{
  		const void* data = buf.readableBegin();
  	  int32_t tmp = *(const int32_t*)(data);
  		int32_t siz = ::ntohl(tmp);
  		if(siz > 65535 || siz < 0)
  		{
        cerr << "Invalid size " << siz << "\n";
  			conn->shutDown();
  		}
  		else if(buf.readableBytes() >= siz + HEAD_SIZE)
  		{
        int siz = 0;
  			string msg;
  		  buf.read(&siz, sizeof(siz));
  		  buf.read(msg);
  		  m_msgCallback(conn, msg, tim);
  		}
  		else
  		{
  			// Wait for more.
  			break;
  		}
  	}
  }
  
  void send(CLTcpConn* conn, const string& msg) 
  {
  	int32_t siz = ::htonl(static_cast<int32_t>(msg.size()));
  	CLBuffer buf;
  	buf.write(&siz, sizeof(siz));
  	buf.write(msg);
  	conn->send(buf.readableBegin(), buf.readableBytes());
  }
  
private:
  static const unsigned int HEAD_SIZE = 4;
  MsgCallback m_msgCallback;
};

class ChatServer
{
public:
  ChatServer()
		: m_tcpServer("localhost", 1238),
		m_mutex(new CLMutex)
	{
		m_codec.reset(new SimpleCodec(bind(&ChatServer::handleMsg, this, _1, _2, _3)));
	  m_tcpServer.setConnCallback(bind(&ChatServer::handleConnection, this, _1));
		m_tcpServer.setMessageCallback(bind(&SimpleCodec::handleMsg, m_codec.get(), _1, _2, _3));
		m_tcpServer.setCloseCallback(bind(&ChatServer::handleClose, this, _1));
	}

	void start(void)
	{
		m_tcpServer.start();
	}

private:
	typedef set<CLTcpConnSPtr> ConnTable;

private:
	void handleConnection(const CLTcpConnSPtr& conn)
	{
    CLLockGuard guard(*m_mutex);
		m_connections.insert(conn);
	}

  void handleMsg(const CLTcpConnSPtr& conn, const string& msg, const CLTimestamp& tim)
	{
    CLLockGuard guard(*m_mutex);
	  ConnTable::iterator itr = m_connections.begin();
	  ConnTable::iterator itrEnd = m_connections.end();
	  for(; itr != itrEnd; ++itr)
		{
		  m_codec->send((*itr).get(), msg);
		}
	}

	void handleClose(const CLTcpConnSPtr& conn)
	{
    m_connections.erase(conn);
	}

private:
	CLTcpServer m_tcpServer;
  unique_ptr<SimpleCodec> m_codec;
	unique_ptr<CLLock> m_mutex;
	ConnTable m_connections;
};

////// Client
class ChatClient
{
public:
  ChatClient()
		: m_tcpClient(new CLTcpClient("localhost", 1238)),
		m_mutex(new CLMutex)
	{
    m_codec.reset(new SimpleCodec(bind(&ChatClient::handleMsg, this, _1, _2, _3)));
	  m_tcpClient->setConnCallback(bind(&ChatClient::handleConnection, this, _1));
	  m_tcpClient->setMessageCallback(bind(&SimpleCodec::handleMsg, m_codec.get(), _1, _2, _3));
	  m_tcpClient->enableRetry();
	}

	void connect(void)
	{
		m_tcpClient->connect();
	}

	void write(const string& msg)
	{
		CLLockGuard guard(*m_mutex);
	  if(m_connection)
		{
			m_codec->send(m_connection.get(), msg);
		}
	}

private:
	void handleConnection(const CLTcpConnSPtr& conn)
	{
		CLLockGuard guard(*m_mutex);
		cout << "new connection\n";
	  m_connection = conn;	
	}

	void handleMsg(const CLTcpConnSPtr& conn,
			           const string& msg,
								 const CLTimestamp& tim)
	{
    printf("%s\n", msg.c_str());
	}

private:
	unique_ptr<CLTcpClient> m_tcpClient;
	unique_ptr<SimpleCodec> m_codec;
  unique_ptr<CLLock> m_mutex;
	CLTcpConnSPtr m_connection;
};

int main(int argc, char* argv[])
{
  if((argc == 2) && argv[1][0] == 's')
	{
    CLSignalHandler sigHandler;
    ChatServer chatServer;
	  chatServer.start();
		sigHandler.waitForSignal();
	}
	else if((argc == 2) && argv[1][0] == 'c')
	{
    ChatClient chatClient;
		chatClient.connect();
		string line;
		while(getline(cin, line))
		{
      string msg(line.c_str());
			chatClient.write(msg);
		}
	}
	else
	{
		cout << "usage: ./test_chatserver s/c\n";
	}

  return 0;
}
